package com.aispeech.jarpc;

import com.aispeech.jarpc.message.ArpcMessage;
import com.aispeech.jarpc.message.EMsgType;
import com.aispeech.jarpc.op.AbstractRequest;
import com.aispeech.jarpc.op.VirtualOp;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.channels.Selector;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class JArpc {
    private final static Map<String, VirtualOp> cachedVirtualOpMap = new ConcurrentHashMap<>();

    private final static ObjectMapper objectMapper = new ObjectMapper();
    private static Selector selector = null;
    private final String host;
    private final int port;
    private int timeout;
    private byte[] sessionId;
    private TcpClient client;
    private IArpcCallback callback;

    public JArpc(String host, int port, int timeout) {
        this(host, port, timeout, null);
    }

    public JArpc(String host, int port, int timeout, IArpcCallback callback) {
        this.host = host;
        this.port = port;
        this.timeout = timeout * 1000;
        this.callback = callback;
    }

    public static void initialize() throws IOException {
        selector = Selector.open();
        new EventLoop(selector).start();
    }

    public static Selector Selector() {
        return selector;
    }

    public byte[] getSessionId() {
        return sessionId;
    }

    public boolean isConnected() {
        return client != null && client.isConnected();
    }

    public String getHost() {
        return host;
    }

    public int getPort() {
        return port;
    }

    public int getTimeout() {
        return timeout;
    }

    public void setTimeout(int msecs) {
        this.timeout = msecs;
    }

    public IArpcCallback getCallback() {
        return callback;
    }

    private void updateSessionId(String sessionId) {
        this.sessionId = new byte[32];
        for (int i = 0; i < Math.min(32, sessionId.length()); i++) {
            this.sessionId[i] = (byte) sessionId.charAt(i);
        }
    }

    public void start(String sessionId, AbstractRequest request) throws IOException, ArpcException {
        updateSessionId(sessionId);

        String opType = request.OpType();
        VirtualOp virtualOp = cachedVirtualOpMap.get(opType);

        String host;
        int port;

        if (virtualOp == null) {
            host = this.host;
            port = this.port;
        } else {
            host = virtualOp.getHost();
            port = virtualOp.getPort();
        }

        client = new TcpClient(this);
        client.connect(host, port);
        client.wait(timeout);

        if (client.getException() != null)
            throw client.getException();

        if (!client.isConnected())
            throw new ArpcException(String.format("connect timeout. host: %s, port: %d", host, port));

        client.sendWithSessionId(request.toJson(), this.sessionId);

        client.wait(timeout);

        if (client.getException() != null)
            throw client.getException();

        ArpcMessage message = client.getMessageRecving();
        if (message == null) {
            throw new ArpcException(EArpcState.S_START_FAILED);
        } else if (message.getMsgType() == EMsgType.REDIRECT) {
            VirtualOp op = objectMapper.readValue(message.getBody(), VirtualOp.class);
            close();

            client = new TcpClient(this);
            client.connect(op.getHost(), op.getPort());
            client.wait(timeout);

            if (client.getException() != null)
                throw client.getException();

            client.sendWithSessionId(request.toJson(), this.sessionId);
            client.wait(timeout);

            if (client.getException() != null)
                throw client.getException();

            message = client.getMessageRecving();
            if (message == null) {
                throw new ArpcException(EArpcState.S_START_FAILED);
            } else {
                JsonNode jsonNode = objectMapper.readTree(message.getBody(), 0, message.getLength());

                int code = jsonNode.get("code").asInt(EArpcState.S_INTERNAL_ERROR.getValue());
                if (code != 0)
                    throw new ArpcException(EArpcState.S_INTERNAL_ERROR.getValue(), jsonNode.toString());
            }

            cachedVirtualOpMap.put(opType, op);

        } else {
            JsonNode jsonNode = objectMapper.readTree(message.getBody(), 0, message.getLength());

            int code = jsonNode.get("code").asInt(EArpcState.S_INTERNAL_ERROR.getValue());
            if (code != 0)
                throw new ArpcException(EArpcState.S_INTERNAL_ERROR.getValue(), jsonNode.toString());
        }
    }

    public ArpcMessage feed(byte[] data, int len) throws IOException, ArpcException {
        client.sendBinary(data, len);

        if (client.getException() != null)
            throw client.getException();

        return client.getMessageRecving();
    }

    public void stop() throws IOException, ArpcException {
        stop(0);
        close();
    }

    public ArpcMessage stop(int timeout) throws IOException, ArpcException {
        client.sendBinary(new byte[0], 0);

        if (timeout > 0) {
            client.wait(timeout * 1000);

            if (client.getException() != null)
                throw client.getException();
        }

        return client.getMessageRecving();
    }

    public ArpcMessage process(String sessionId, AbstractRequest request) throws ArpcException {
        updateSessionId(sessionId);

        String opType = request.OpType();
        VirtualOp virtualOp = cachedVirtualOpMap.get(opType);

        String host;
        int port;

        if (virtualOp == null) {
            host = this.host;
            port = this.port;
        } else {
            host = virtualOp.getHost();
            port = virtualOp.getPort();
        }

        log.info("[ process ] sessionId: {} op: {}, begin connecting to: {}:{}", sessionId, opType, host, port);

        SyncSocket syncSocket = new SyncSocket();

        try {
            syncSocket.connect(host, port, timeout);

            if (!syncSocket.isConnected())
                throw new ArpcException(String.format("connect timeout or not connected. host: %s, port: %d", host, port));

            syncSocket.send(this.sessionId, request.toJsonBin().asBuffer());
            log.info("[ process ] sessionId: {} op: {}, begin send request to: {}:{}", sessionId, opType, host, port);
//            log.info("[ process ] sessionId: {} op: {}, begin send request: {} to: {}:{}", sessionId, opType, request.toJson(), host, port);

            ArpcMessage message = syncSocket.receive();

            if (message.getMsgType() == EMsgType.REDIRECT) {
                VirtualOp op = objectMapper.readValue(message.getBody(), VirtualOp.class);
                host = op.getHost();
                port = op.getPort();
                syncSocket.close();
                syncSocket.connect(host, port, timeout);

                syncSocket.send(this.sessionId, request.toJsonBin().asBuffer());
                cachedVirtualOpMap.put(opType, op);

                message = syncSocket.receive();
            }

            if (message == null)
                throw new ArpcException(String.format("failed parse message. session: %s, op: %s, host: %s, port: %d", sessionId, opType, host, port));

            return message;
        } catch (IOException ex) {
            log.error(String.format("[ aprc.process ] session: %s, op: %s, host: %s, port: %d", sessionId, opType, host, port), ex);
            throw new ArpcException(ex.getMessage());
        } finally {
            if (syncSocket != null) {
                syncSocket.close();
            }
        }
    }

    public void close() {
        if (client != null) {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }

            client = null;
        }
    }
}
